package org.rcy.framework.mq.listener;

import org.rcy.framework.api.mq.AbstractMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;

import java.lang.reflect.ParameterizedType;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

public class RabbitMqMessageListenerRegistrar implements RabbitMessageListenerRegistrar, BeanFactoryAware, DisposableBean {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    private BeanFactory beanFactory;
    private final AtomicInteger counter = new AtomicInteger();
    private final Map<String, MessageListenerContainer> listenerContainers = new ConcurrentHashMap();

    public RabbitMqMessageListenerRegistrar() {
    }

    public <M extends AbstractMessage> void register(RabbitMessageListener<M> listener) {
        ParameterizedType type = (ParameterizedType)listener.getClass().getGenericSuperclass();
        Class<?> clazz = (Class)type.getActualTypeArguments()[0];
        String queueName = clazz.getName();
        ((RabbitAdmin)this.beanFactory.getBean(RabbitAdmin.class)).declareQueue(new Queue(queueName));
        MessageListenerAdapter adapter = new MessageListenerAdapter(listener);
        adapter.setMessageConverter((MessageConverter)this.beanFactory.getBean(MessageConverter.class));
        adapter.setDefaultListenerMethod("onMessage");
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setQueueNames(new String[]{queueName});
        container.setConnectionFactory((ConnectionFactory)this.beanFactory.getBean(ConnectionFactory.class));
        container.setMessageListener(adapter);
        container.start();
        this.listenerContainers.put(container.getClass().getSimpleName() + this.counter.getAndIncrement(), container);
    }

    public void destroy() {
        Iterator var1 = this.getListenerContainers().iterator();

        while(var1.hasNext()) {
            MessageListenerContainer listenerContainer = (MessageListenerContainer)var1.next();
            if (listenerContainer instanceof DisposableBean) {
                try {
                    ((DisposableBean)listenerContainer).destroy();
                } catch (Exception var4) {
                    this.logger.warn("Failed to destroy listener container [" + listenerContainer + "]", var4);
                }
            }
        }

    }

    public Collection<MessageListenerContainer> getListenerContainers() {
        return Collections.unmodifiableCollection(this.listenerContainers.values());
    }

    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }
}
